package com.xsomnus.open.zookeeper.controller;

import com.xsomnus.open.zookeeper.common.Const;
import com.xsomnus.open.zookeeper.config.ZkClient;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.framework.recipes.locks.InterProcessReadWriteLock;
import org.apache.curator.framework.recipes.shared.SharedCount;
import org.apache.curator.framework.recipes.shared.VersionedValue;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;

import java.util.*;
import java.util.concurrent.*;
import java.util.function.Consumer;

/**
 * @author @xsomnus666_xiawenye★
 * @since 2019/4/18 0018 14:02
 * 当你发现自己的才华撑不起野心时，就请安静下来学习吧
 */
@Slf4j
@RestController
@RequestMapping("/zk")
public class ZkController {

    @Autowired
    ZkClient zkClient;


    @GetMapping("/test")
    public String test() {
        return RandomStringUtils.randomAlphabetic(8, 16);
    }

    /**
     * 添加节点
     */
    @PostMapping("/{type}/{znode}")
    private String create(@PathVariable Integer type, @PathVariable String znode, @RequestBody String nodeData) {
        znode = Const.ZNODE_PREFIX + znode;
        try {
            zkClient.createNode(CreateMode.fromFlag(type), znode, nodeData);
        } catch (KeeperException e) {
            e.printStackTrace();
        }
        return znode;
    }


    /**
     * 更新节点信息
     */
    @PutMapping("/{znode}")
    private boolean update(@PathVariable String znode, @RequestBody String nodeData) {
        znode = Const.ZNODE_PREFIX + znode;
        zkClient.setNodeData(znode, nodeData.getBytes());
        return true;
    }


    /**
     * 删除节点
     */
    @DeleteMapping("/{znode}")
    private boolean delete(@PathVariable String znode) {
        znode = Const.ZNODE_PREFIX + znode;
        zkClient.deleteNode(znode);
        return true;
    }

    /**
     * 查找节点的内容
     */
    @GetMapping("/{znode}")
    public String find(@PathVariable String znode) {
        return new String(zkClient.getNodeData(Const.ZNODE_PREFIX + znode));
    }


    /**
     * 读写锁
     * <p>
     * 写锁跟读锁互斥,读锁跟读锁共享
     */
    @GetMapping("/writeLock")
    public String readLock(@RequestParam Integer lockType, @RequestParam String znode) {
        String requestId = UUID.randomUUID().toString();
        znode = Const.ZNODE_PREFIX + znode;

        InterProcessReadWriteLock readWriteLock = zkClient.getReadWriteLock(znode);
        InterProcessMutex writeLock = readWriteLock.writeLock();
        InterProcessMutex readLock = readWriteLock.readLock();

        Runnable writeTask = () -> {
            try {
                log.debug("------------- {} write lock -------------", requestId);
                writeLock.acquire();
                log.debug("---------- {} write lock acquire --------", requestId);
                Thread.sleep(10_000);
                writeLock.release();
                log.debug("---------- {} write release -------------", requestId);
            } catch (Exception e) {
                e.printStackTrace();
            }
        };


        Runnable readTask = () -> {
            try {
                log.debug("------------- read lock -------------");
                readLock.acquire();
                log.debug("---------- read lock acquire --------");
                Thread.sleep(5_000);
                log.debug("---------- read release -------------");
                readLock.release();
            } catch (Exception e) {
                e.printStackTrace();
            }
        };

        if (lockType == 0) {
            new Thread(writeTask).start();
        } else if (lockType == 1) {
            new Thread(readTask).start();
        }

        return "success";

    }


    /**
     * 监听节点
     * 监控整个树上的所有节点
     */
    @PostMapping("/watchPath")
    public String watchPath(@RequestParam String znode) {
        znode = Const.ZNODE_PREFIX + znode;

        zkClient.watchPath(znode, (client, event) -> {
            log.info("event:" + event.getType() + " |path:" + (null != event.getData() ? event.getData().getPath() : null));
            if (event.getData() != null && event.getData().getData() != null) {
                log.info("发生变化的节点内容为：" + new String(event.getData().getData()));
            }
        });
        return "success";
    }


    /**
     * 模拟分布式计数器
     */
    @PostMapping("/v1/counter")
    public String counter(@RequestParam String znode) {
        znode = Const.ZNODE_PREFIX + znode;
        SharedCount sharedCount = new SharedCount(zkClient.getClient(), znode, 0);
        try {
            sharedCount.start();

            // 模拟线程池
            ExecutorService executorService = Executors.newCachedThreadPool();
            Consumer<SharedCount> consumer = sharedCount1 -> {
                try {
                    List<Callable<Boolean>> callableList = new ArrayList<>();

                    Callable<Boolean> call = () -> {
                        boolean result = false;
                        try {
                            long time = System.currentTimeMillis();
                            while (!result) {
                                VersionedValue<Integer> oldVersion = sharedCount.getVersionedValue();
                                int newVersion = oldVersion.getValue() + 1;
                                result = sharedCount.trySetCount(oldVersion, newVersion);
                                if (System.currentTimeMillis() - time > 10_000 || result) {
                                    break;
                                }
                                TimeUnit.MICROSECONDS.sleep(new Random().nextInt(100) + 1);
                            }
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                        return result;
                    };

                    for (int i = 0; i < 1000; i++) {
                        callableList.add(call);
                    }
                    List<Future<Boolean>> futures = executorService.invokeAll(callableList);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            };
            consumer.accept(sharedCount);
            log.debug("最后的计数结果是: " + sharedCount.getCount());
        } catch (Exception e) {
            e.printStackTrace();
        }
        return "success: " + sharedCount.getCount();
    }


    /**
     * zookeeper生成分布式id
     * @param count 生成id的数量
     *
     */
    @GetMapping("/dui")
    public Collection<String> distributedUniqueID(@RequestParam Integer count) {
        List<String> idList = new ArrayList<>(count);
        for (int i = 0; i < count; i++) {
            idList.add(zkClient.generateId("order"));
        }
        return idList;
    }

}
